package com.atguigu.dga.governance.service.impl;

import com.atguigu.dga.dolphinscheduler.bean.TDsTaskDefinition;
import com.atguigu.dga.dolphinscheduler.bean.TDsTaskInstance;
import com.atguigu.dga.dolphinscheduler.service.TDsTaskDefinitionService;
import com.atguigu.dga.dolphinscheduler.service.TDsTaskInstanceService;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import com.atguigu.dga.governance.mapper.GovernanceAssessDetailMapper;
import com.atguigu.dga.governance.service.GovernanceAssessDetailService;
import com.atguigu.dga.governance.service.GovernanceMetricService;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.bean.TableMetaInfoExtra;
import com.atguigu.dga.meta.mapper.TableMetaInfoMapper;
import com.atguigu.dga.meta.service.TableMetaInfoExtraService;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.atguigu.dga.util.SpringBeanProvider;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.base.CaseFormat;
import jdk.nashorn.internal.ir.Block;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.lang.reflect.InvocationTargetException;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * <p>
 * 治理考评结果明细 服务实现类
 * </p>
 *
 * @author tubage
 * @since 2024-03-16
 */
@Service
@DS("dga")
public class GovernanceAssessDetailServiceImpl extends ServiceImpl<GovernanceAssessDetailMapper, GovernanceAssessDetail> implements GovernanceAssessDetailService {
    @Autowired
    TableMetaInfoService tableMetaInfoService;
    @Autowired
    TableMetaInfoExtraService tableMetaInfoExtraService;
    @Autowired
    GovernanceMetricService governanceMetricService;
    @Autowired
    TableMetaInfoMapper tableMetaInfoMapper;
    @Autowired
    SpringBeanProvider springBeanProvider;
    @Autowired
    TDsTaskDefinitionService tDsTaskDefinitionService;
    @Autowired
    TDsTaskInstanceService tDsTaskInstanceService;
    /**
     * 考评
     * 1 获取所有表
     * 2 获取所有指标
     * 3 每张表每个指标循环迭代逐一考评
     * 4 批量写入表中
     * @param assessDate
     */
    @Override
    public void mainAssess(String assessDate) throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
        // TODO 0 幂等性
        remove(new QueryWrapper<GovernanceAssessDetail>()
                .eq("assess_date", assessDate));
        // TODO 1 获取所有待考评表
        // 方案一 直接用MP 将tableMetaInfo全部查出来、迭代、再查询tableMetaInfoExtra
        // 不推荐 循环中与MySQL交互
        //  List<TableMetaInfo> tableMetaInfoList1 = tableMetaInfoService.list(
        //          new QueryWrapper<TableMetaInfo>()
        //                  .eq("assess_date", assessDate)
        //  );
        //  for (TableMetaInfo tableMetaInfo : tableMetaInfoList1) {
        //      TableMetaInfoExtra tableMetaInfoExtra = tableMetaInfoExtraService.getOne(
        //              new QueryWrapper<TableMetaInfoExtra>()
        //                      .eq("schema_name", tableMetaInfo.getSchemaName())
        //                      .eq("table_name", tableMetaInfo.getTableName())
        //      );
        //      tableMetaInfo.setTableMetaInfoExtra(tableMetaInfoExtra);
        //        }

        // 方案二 使用MP 将tableMetaInfo tableMetaInfoExtra 都拿出来 内存中关联
        //  extra遍历转 map结构  循环metaInfo 通过map get指定key 设置extra
        //  List<TableMetaInfo> tableMetaInfoList1 = tableMetaInfoService.list(
        //          new QueryWrapper<TableMetaInfo>()
        //                  .eq("assess_date", assessDate)
        //  );
        //  List<TableMetaInfoExtra> tableMetaInfoExtraList = tableMetaInfoExtraService.list();
        //  Map<String, TableMetaInfoExtra> tableMetaInfoExtraMap = new HashMap<>(tableMetaInfoExtraList.size());
        //  for (TableMetaInfoExtra tableMetaInfoExtra : tableMetaInfoExtraList) {
        //      tableMetaInfoExtraMap.put(tableMetaInfoExtra.getSchemaName() + ":" + tableMetaInfoExtra.getTableName(),
        //              tableMetaInfoExtra);
        //  }
        //  for (TableMetaInfo tableMetaInfo : tableMetaInfoList1) {
        //      tableMetaInfo.setTableMetaInfoExtra(tableMetaInfoExtraMap.get(tableMetaInfo.getSchemaName() +":" + tableMetaInfo.getTableName()));
        //  }



        // 方案三 join 手动封装tableMetaInfo和tableMetaInfoExtra 在tableMetaInfo 写Select mapper
        // mapper xml放resource  关联@ResultMap  改映射 id auto extra
        // application.properties配置 mybatis mapper xml文件位置
        List<TableMetaInfo> tableMetaInfoList = tableMetaInfoMapper.selectAllTableMetaInfos(assessDate);
//        System.out.println(tableMetaInfoList);



        // TODO 2 获取所有指标 metric
        List<GovernanceMetric> governanceMetricList = governanceMetricService.list(
                new QueryWrapper<GovernanceMetric>()
                        .eq("is_disabled", "0")
        );
//        System.out.println("governanceMetricList = " + governanceMetricList);

        // TODO 5 补充DS ： 获取DS中任务定义  转map结构 get获取封装进参数里
        List<TDsTaskDefinition> tDsTaskDefinitionList = tDsTaskDefinitionService.getTDsTaskDefinitionList();
//        System.out.println(tDsTaskDefinitionList);
        Map<String, TDsTaskDefinition> tDsTaskDefinitionMap = new HashMap<>(tDsTaskDefinitionList.size());
        for (TDsTaskDefinition tDsTaskDefinition : tDsTaskDefinitionList) {
            tDsTaskDefinitionMap.put(tDsTaskDefinition.getName(), tDsTaskDefinition);
        }
        // TODO 5 补充DS ：任务实例  转map结构 get获取封装进参数里
        List<TDsTaskInstance> tdsTaskInstanceList = tDsTaskInstanceService.getTdsTaskInstanceListByIn(assessDate);
//        System.out.println(tdsTaskInstanceList);
        Map<String, TDsTaskInstance> tDsTaskInstanceMap = new HashMap<>();
        for (TDsTaskInstance tDsTaskInstance : tdsTaskInstanceList) {
            tDsTaskInstanceMap.put(tDsTaskInstance.getName(), tDsTaskInstance);
        }


        // TODO 3 逐一考评
        // 1) 指标描述落实成可执行功能 一个指标一个类一个方法 考评器Assessor 开闭原则
        // 2) 如何通过指标确定对应的考评器？动态

        // 循环表 内循环考评
        // 方案一：约定+反射：基于约定好的类名 通过反射的方式获取对应考评器对象
        // 约定：考评器的类名以 指标的编码的驼峰形式开头 以assessor结尾
        // 约定：考评器的包以 指标的类型来命名
        // for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
        //     for (GovernanceMetric governanceMetric : governanceMetricList) {
        //         String basePackage = "com.atguigu.dga.governance.assessor";
        //         String governanceMetricPackage = governanceMetric.getGovernanceType().toLowerCase();
        //         String metricCode = governanceMetric.getMetricCode();
        //         String className = CaseFormat.UPPER_UNDERSCORE.to(CaseFormat.UPPER_CAMEL, metricCode) + "Assessor";
        //         String fullClassName = basePackage + "." + governanceMetricPackage + "." + className ;
        //
        //         Class<?> clsObj = Class.forName(fullClassName);
        //         Assessor assessor = (Assessor)clsObj.getDeclaredConstructor().newInstance();
        //         assessor.doAssess();
        //     }
        // }

        // 方案二：约定+Spring容器：考评器组件名使用对应指标编码， 考评器管理到容器中 @Component("")
        // 获取指标编码，到容器里找对应的component 需要把容器拿到手里面 写一个工具类
        // util SpringBean SpringBeanProvider 泛型

        long start = System.currentTimeMillis();

        // TODO 优化2 多线程
        ArrayList<CompletableFuture<GovernanceAssessDetail>> futures =
                new ArrayList<>(tableMetaInfoList.size() * governanceMetricList.size());
        // 线程池
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(4, 6, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1500));
//        ArrayList<GovernanceAssessDetail> governanceAssessDetails = new ArrayList<>(tableMetaInfoList.size() *governanceMetricList.size() );

        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
            for (GovernanceMetric governanceMetric : governanceMetricList) {
                // TODO  优化1 白名单 判断是否跳过该表考评
                String skipAssessTables = governanceMetric.getSkipAssessTables();
                if(skipAssessTables != null && !skipAssessTables.trim().isEmpty()) {
                    String[] skipAssessTableArr = skipAssessTables.split(",");
                    List<String> skipAssessTableList = Arrays.asList(skipAssessTableArr);
                    if(skipAssessTableList.contains(tableMetaInfo.getTableName())) {
                        continue;
                    }
                }

                String metricCode = governanceMetric.getMetricCode();
                Assessor assessor = springBeanProvider.getBean(metricCode, Assessor.class);
                // 考评参数
                String mapKey = tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName();
                AssessParam assessParam = AssessParam.builder()
                        .governanceMetric(governanceMetric)
                        .tableMetaInfo(tableMetaInfo)
                        .assessDate(assessDate)
                        .tableMetaInfoList(tableMetaInfoList)
                        .tDsTaskDefinition(tDsTaskDefinitionMap.get(mapKey)) // TODO 6 补充考评参数
                        .tDsTaskInstance(tDsTaskInstanceMap.get(mapKey))
                                .build();

//                GovernanceAssessDetail governanceAssessDetail = assessor.doAssess(assessParam);
//                governanceAssessDetails.add(governanceAssessDetail);

                // TODO 优化2 多线程
                CompletableFuture<GovernanceAssessDetail> future = CompletableFuture.supplyAsync(
                        () -> assessor.doAssess(assessParam),
                        poolExecutor
                );
                futures.add(future);
            }
        }
        List<GovernanceAssessDetail> governanceAssessDetails = futures.stream().map(CompletableFuture::join).collect(Collectors.toList());

        // 优化考评器Assessor  final  void doAssess()定义好流程

        // 封装传参对象 考评参数

        long end = System.currentTimeMillis();
        System.out.println("📢考评耗时：" + (end -  start));

        // TODO 4 保存
        saveBatch(governanceAssessDetails);


    }

    @Override
    public List<GovernanceAssessDetail> getProblemListByTypeAndPage(String governType, Integer pageNo, Integer pageSize) {
        // TODO 问题项接口
        int start = (pageNo - 1) * pageSize;
        return list(
                new QueryWrapper<GovernanceAssessDetail>()
                        .eq("governance_type",governType)
                        .lt("assess_score", 10)
                        .inSql("assess_date","select max(assess_date) from governance_assess_detail")
                        .last("limit " + start + "," + pageSize)
        );

    }
}
